package index;

import heap.*;
import exceptions.*;
import Common.GlobalConst;
import Common.GoggleDB;
import Transaction.*;
import java.io.*;

import storage.data.Page;
import storage.data.PageID;

public class BTFile extends IndexFile implements GlobalConst {

	private final static int MAGIC0 = 1989;

	private final static String lineSep = System.getProperty("line.separator");

	private static FileOutputStream fos;

	private static DataOutputStream trace;

	private BTHeaderPage headerPage;

	private PageID headerPageId;

	private String dbname;

	public static void traceFilename(String filename) throws IOException {

		//fos = new FileOutputStream(filename);
		// trace = new DataOutputStream(fos);
	}

	public static void destroyTrace() throws IOException {
		if (trace != null)
			trace.close();
		if (fos != null)
			fos.close();
		fos = null;
		trace = null;
	}

	public BTHeaderPage getHeaderPage() {
		return headerPage;
	}

	private PageID getFileEntry(String filename) throws Exception {
		return GoggleDB.storageInterface.get_db_by_name(filename);

	}

	private Page pinPage(PageID pageno) throws Exception {
		Page page = new Page();
		GoggleDB.bufferMgr.pinPage(pageno, page, false/* Rdisk */);
		return page;
	}

	private PageID addFileEntry(String fileName, Transaction tx)
			throws Exception {
		return GoggleDB.storageInterface.add_db(tx, fileName);
	}

	private void unpinPage(PageID pageno) throws Exception {
		GoggleDB.bufferMgr.unpinPage(pageno, false /* = not DIRTY */);

	}

	private void deleteFileEntry(String filename, Transaction tx)
			throws Exception {
		GoggleDB.storageInterface.del_db(tx, filename);
	}

	private void unpinPage(PageID pageno, boolean dirty) throws Exception {
		GoggleDB.bufferMgr.unpinPage(pageno, dirty);
	}

	private void freePage(PageID pageno) throws Exception {
		GoggleDB.bufferMgr.deallocatePage(pageno, 1);
	}

	/**
	 * an index file with given filename should already exist;
	 * 
	 * @param filename
	 * @throws Exception
	 */
	public BTFile(String filename) throws Exception {

		headerPageId = getFileEntry(filename);

		headerPage = new BTHeaderPage(headerPageId);
		dbname = new String(filename);
		/*
		 * headerPageId is the PageId of this BTreeFile's header page; -
		 * headerPage, headerPageId valid and pinned - dbname contains a copy of
		 * the name of the database
		 */
	}

	public BTFile(String filename, int keytype, int keysize,
			int delete_fashion, Transaction tx) throws Exception {
		boolean notFound = false;
		//System.out.println(filename);
		try {
			headerPageId = getFileEntry(filename);
		} catch (Exception e) {
			notFound = true;
		}
		if (notFound) {
			headerPage = new BTHeaderPage(filename, tx);
			headerPageId = headerPage.getPageID(tx, filename);
			// headerPageId=addFileEntry(filename);
			headerPage.setMagic0(MAGIC0, tx, filename);
			headerPage.setRootId(new PageID(INVALID), tx, filename);
			headerPage.setKeyType((short) keytype, tx, filename);
			headerPage.setMaxKeySize(keysize, tx, filename);
			headerPage.setDeleteFashion(delete_fashion, tx, filename);
			headerPage.setType(NodeType.BTHEAD, tx, filename, headerPageId);
			// System.out.println("header"+headerPage.curPage.pageId);
		} else {
			headerPage = new BTHeaderPage(headerPageId);
		}

		dbname = new String(filename);

	}

	public void close() throws Exception {
		if (headerPage != null) {
			GoggleDB.bufferMgr.unpinPage(headerPageId, true);
			headerPage = null;
		}
	}

	public void destroyFile(Transaction tx, String table) throws Exception {
		if (headerPage != null) {
			PageID pgId = headerPage.getRootId(tx, table);
			if (pgId.pageId != INVALID)
				deleteFile(pgId, tx, table);
			unpinPage(headerPageId);
			freePage(headerPageId);
			deleteFileEntry(dbname, tx);
			headerPage = null;
		}
	}

	private void deleteFile(PageID pageno, Transaction tx, String table)
			throws Exception {
		BTPage sortedPage;
		Page page = pinPage(pageno);
		sortedPage = new BTPage(page, headerPage.getKeyType(tx, table));

		if (sortedPage.getType(tx, table) == NodeType.INDEX) {
			IndexPage indexPage = new IndexPage(page, headerPage.getKeyType(tx,
					table), tx, table);
			RID rid = new RID();
			PageID childId;
			KeyDataEntry entry;
			for (entry = indexPage.getFirst(rid, tx, table); entry != null; entry = indexPage
					.getNext(rid, tx, table)) {
				childId = ((IndexData) (entry.data)).getData();
				deleteFile(childId, tx, table);
			}
		} else { // BTLeafPage
			unpinPage(pageno);
			freePage(pageno);
		}
	}

	private void updateHeader(PageID newRoot, Transaction tx, String table)
			throws Exception {

		BTHeaderPage header;
		PageID oldData;

		header = new BTHeaderPage(pinPage(headerPageId));
		oldData = headerPage.getRootId(tx, table);
		header.setRootId(newRoot, tx, table);
		// clock in dirty bit to bm so our dtor needn't have to worry about it
		unpinPage(headerPageId, true /* = DIRTY */);
		// ASSERTIONS:
		// - headerPage, headerPageId valid, pinned and marked as dirty

	}

	public void insert(KeyClass key, RID rid, Transaction tx, String table)
			throws Exception {
		KeyDataEntry newRootEntry;

		// TWO CASES:
		// 1. headerPage.root == INVALID_PAGE:
		// - the tree is empty and we have to create a new first page;
		// this page will be a leaf page
		// 2. headerPage.root != INVALID_PAGE:
		// - we call _insert() to insert the pair (key, rid)

		if (trace != null) {
			trace.writeBytes("INSERT " + rid.pageNo + " " + rid.slotNo + " "
					+ key + lineSep);
			trace.writeBytes("DO" + lineSep);
			trace.flush();
		}

		if (headerPage.getRootId(tx, table).pageId == INVALID) {
			PageID newRootPageId;
			LeafPage newRootPage;
			RID dummyrid;

			newRootPage = new LeafPage(headerPage.getKeyType(tx, table), tx,
					table);
			newRootPageId = newRootPage.getCurPage(tx, table);

			if (trace != null) {
				trace.writeBytes("NEWROOT " + newRootPageId + lineSep);
				trace.flush();
			}

			newRootPage.setNextPage(new PageID(INVALID), tx, table);
			newRootPage.setPrevPage(new PageID(INVALID), tx, table);

			// ASSERTIONS:
			// - newRootPage, newRootPageId valid and pinned

			newRootPage.insertRecord(key, rid, tx, table);

			if (trace != null) {
				trace.writeBytes("PUTIN node " + newRootPageId + lineSep);
				trace.flush();
			}
			unpinPage(newRootPageId, true); /* = DIRTY */
			updateHeader(newRootPageId, tx, table);
			if (trace != null) {
				trace.writeBytes("DONE" + lineSep);
				trace.flush();
			}
			return;
		}

		// ASSERTIONS:
		// - headerPageId, headerPage valid and pinned
		// - headerPage.root holds the pageId of the root of the B-tree
		// - none of the pages of the tree is pinned yet

		if (trace != null) {
			trace.writeBytes("SEARCH" + lineSep);
			trace.flush();
		}

		// System.out.println("key:"+key+" rid: "+rid.slotNo+"
		// "+rid.pageNo.pageId);
		newRootEntry = insertsp(key, rid, headerPage.getRootId(tx, table), tx,
				table);

		// TWO CASES:
		// - newRootEntry != null: a leaf split propagated up to the root
		// and the root split: the new pageNo is in
		// newChildEntry.data.pageNo
		// - newRootEntry == null: no new root was created;
		// information on headerpage is still valid

		// ASSERTIONS:
		// - no page pinned

		if (newRootEntry != null) {
			IndexPage newRootPage;
			PageID newRootPageId;
			Object newEntryKey;
			// the information about the pair <key, PageId> is
			// packed in newRootEntry: extract it

			newRootPage = new IndexPage(headerPage.getKeyType(tx, table), tx,
					table);
			newRootPageId = newRootPage.getCurPage(tx, table);

			// ASSERTIONS:
			// - newRootPage, newRootPageId valid and pinned
			// - newEntryKey, newEntryPage contain the data for the new entry
			// which was given up from the level down in the recursion

			if (trace != null) {
				trace.writeBytes("NEWROOT " + newRootPageId + lineSep);
				trace.flush();
			}

			newRootPage.insertKey(newRootEntry.key,
					((IndexData) newRootEntry.data).getData(), tx, table);

			// the old root split and is now the left child of the new root
			newRootPage.setPrevPage(headerPage.getRootId(tx, table), tx, table);
			unpinPage(newRootPageId, true /* = DIRTY */);
			updateHeader(newRootPageId, tx, table);
		}

		if (trace != null) {
			trace.writeBytes("DONE" + lineSep);
			trace.flush();
		}
		return;
	}

	private KeyDataEntry insertsp(KeyClass key, RID rid, PageID currentPageId,
			Transaction tx, String table) throws Exception {

		BTPage currentPage;
		Page page;
		KeyDataEntry upEntry;

		page = pinPage(currentPageId);
		currentPage = new BTPage(page, headerPage.getKeyType(tx, table));

		if (trace != null) {
			trace.writeBytes("VISIT node " + currentPageId + lineSep);
			trace.flush();
		}
		// TWO CASES:
		// - pageType == INDEX:
		// recurse and then split if necessary
		// - pageType == LEAF:
		// try to insert pair (key, rid), maybe split

		if (currentPage.getType(tx, table) == NodeType.INDEX) {
			IndexPage currentIndexPage = new IndexPage(page, headerPage
					.getKeyType(tx, table), tx, table);
			PageID currentIndexPageId = currentPageId;
			PageID nextPageId;

			nextPageId = currentIndexPage.getPageNoByKey(key, tx, table);
			// now unpin the page, recurse and then pin it again
			unpinPage(currentIndexPageId);

			upEntry = insertsp(key, rid, nextPageId, tx, table);
			// two cases:
			// - upEntry == null: one level lower no split has occurred:
			// we are done.
			// - upEntry != null: one of the children has split and
			// upEntry is the new data entry which has
			// to be inserted on this index page

			if (upEntry == null)
				return null;

			currentIndexPage = new IndexPage(pinPage(currentPageId), headerPage
					.getKeyType(tx, table), tx, table);

			// ASSERTIONS:
			// - upEntry != null
			// - currentIndexPage, currentIndexPageId valid and pinned

			// the information about the pair <key, PageId> is
			// packed in upEntry

			// check whether there can still be entries inserted on that page
			if (currentIndexPage.availableSpace(tx, table) >= BT
					.getKeyDataLength(upEntry.key, NodeType.INDEX, tx, table)) {
				// no split has occurred
				currentIndexPage.insertKey(upEntry.key,
						((IndexData) upEntry.data).getData(), tx, table);
				unpinPage(currentIndexPageId, true /* DIRTY */);
				return null;
			}

			// ASSERTIONS:
			// - on the current index page is not enough space available .
			// it splits

			// therefore we have to allocate a new index page and we will
			// distribute the entries
			// - currentIndexPage, currentIndexPageId valid and pinned

			IndexPage newIndexPage;
			PageID newIndexPageId;
			// we have to allocate a new INDEX page and
			// to redistribute the index entries
			newIndexPage = new IndexPage(headerPage.getKeyType(tx, table), tx,
					table);
			newIndexPageId = newIndexPage.getCurPage(tx, table);

			if (trace != null) {
				if (headerPage.getRootId(tx, table).pageId != currentIndexPageId.pageId)
					trace.writeBytes("SPLIT node " + currentIndexPageId
							+ " IN nodes " + currentIndexPageId + " "
							+ newIndexPageId + lineSep);
				else
					trace.writeBytes("ROOTSPLIT IN nodes " + currentIndexPageId
							+ " " + newIndexPageId + lineSep);
				trace.flush();
			}
			// ASSERTIONS:
			// - newIndexPage, newIndexPageId valid and pinned
			// - currentIndexPage, currentIndexPageId valid and pinned
			// - upEntry containing (Key, Page) for the new entry which was
			// given up from the level down in the recursion

			KeyDataEntry tmpEntry;
			PageID tmpPageId;
			RID insertRid;
			RID delRid = new RID();

			for (tmpEntry = currentIndexPage.getFirst(delRid, tx, table); tmpEntry != null; tmpEntry = currentIndexPage
					.getFirst(delRid, tx, table)) {
				newIndexPage.insertKey(tmpEntry.key,
						((IndexData) tmpEntry.data).getData(), tx, table);
				currentIndexPage.deleteSortedRecord(delRid, tx, table);
			}

			// ASSERTIONS:
			// - currentIndexPage empty
			// - newIndexPage holds all former records from currentIndexPage

			// we will try to make an equal split
			RID firstRid = new RID();
			KeyDataEntry undoEntry = null;
			for (tmpEntry = newIndexPage.getFirst(firstRid, tx, table); (currentIndexPage
					.availableSpace(tx, table) > newIndexPage.availableSpace(
					tx, table)); tmpEntry = newIndexPage.getFirst(firstRid, tx,
					table)) {
				// now insert the <key,pageId> pair on the new
				// index page
				undoEntry = tmpEntry;
				currentIndexPage.insertKey(tmpEntry.key,
						((IndexData) tmpEntry.data).getData(), tx, table);
				newIndexPage.deleteSortedRecord(firstRid, tx, table);
			}

			// undo the final record
			if (currentIndexPage.availableSpace(tx, table) < newIndexPage
					.availableSpace(tx, table)) {

				newIndexPage.insertKey(undoEntry.key,
						((IndexData) undoEntry.data).getData(), tx, table);

				currentIndexPage.deleteSortedRecord(new RID(currentIndexPage
						.getCurPage(tx, table), (int) currentIndexPage
						.getSlotCnt(tx, table) - 1), tx, table);
			}

			// check whether <newKey, newIndexPageId>
			// will be inserted
			// on the newly allocated or on the old index page

			tmpEntry = newIndexPage.getFirst(firstRid, tx, table);

			if (BT.KeyCompare(upEntry.key, tmpEntry.key) >= 0) {
				// the new data entry belongs on the new index page
				newIndexPage.insertKey(upEntry.key, ((IndexData) upEntry.data)
						.getData(), tx, table);
			} else {
				currentIndexPage.insertKey(upEntry.key,
						((IndexData) upEntry.data).getData(), tx, table);

				int i = (int) currentIndexPage.getSlotCnt(tx, table) - 1;
				tmpEntry = BT.getEntryFromBytes(currentIndexPage.getData(),
						currentIndexPage.getSlotOffset(i, tx, table),
						currentIndexPage.getSlotLength(i, tx, table),
						headerPage.getKeyType(tx, table), NodeType.INDEX, tx,
						table);

				newIndexPage.insertKey(tmpEntry.key,
						((IndexData) tmpEntry.data).getData(), tx, table);

				currentIndexPage.deleteSortedRecord(new RID(currentIndexPage
						.getCurPage(tx, table), i), tx, table);

			}

			unpinPage(currentIndexPageId, true /* dirty */);

			// fill upEntry
			upEntry = newIndexPage.getFirst(delRid, tx, table);

			// now set prevPageId of the newIndexPage to the pageId
			// of the deleted entry:
			newIndexPage.setPrevPage(((IndexData) upEntry.data).getData(), tx,
					table);

			// delete first record on new index page since it is given up
			newIndexPage.deleteSortedRecord(delRid, tx, table);

			unpinPage(newIndexPageId, true /* dirty */);

			if (trace != null) {
				trace_children(currentIndexPageId, tx, table);
				trace_children(newIndexPageId, tx, table);
			}

			((IndexData) upEntry.data).setData(newIndexPageId);

			return upEntry;
			// ASSERTIONS:
			// - no pages pinned
			// - upEntry holds the pointer to the KeyDataEntry which is
			// to be inserted on the index page one level up
		}

		else if (currentPage.getType(tx, table) == NodeType.LEAF) {
			LeafPage currentLeafPage = new LeafPage(page, headerPage
					.getKeyType(tx, table), tx, table);

			PageID currentLeafPageId = currentPageId;

			// ASSERTIONS:
			// - currentLeafPage, currentLeafPageId valid and pinned

			// check whether there can still be entries inserted on that page
			if (currentLeafPage.availableSpace(tx, table) >= BT
					.getKeyDataLength(key, NodeType.LEAF, tx, table)) {
				// no split has occurred

				currentLeafPage.insertRecord(key, rid, tx, table);

				unpinPage(currentLeafPageId, true /* DIRTY */);

				if (trace != null) {
					trace.writeBytes("PUTIN node " + currentLeafPageId
							+ lineSep);
					trace.flush();
				}

				return null;
			}

			// ASSERTIONS:
			// - on the current leaf page is not enough space available.
			// It splits.
			// - therefore we have to allocate a new leaf page and we will
			// - distribute the entries

			LeafPage newLeafPage;
			PageID newLeafPageId;
			// we have to allocate a new LEAF page and
			// to redistribute the data entries entries
			newLeafPage = new LeafPage(headerPage.getKeyType(tx, table), tx,
					table);
			newLeafPageId = newLeafPage.getCurPage(tx, table);

			newLeafPage.setNextPage(currentLeafPage.getNextPage(tx, table), tx,
					table);
			newLeafPage.setPrevPage(currentLeafPageId, tx, table); // for
																	// dbl-linked
																	// list
			currentLeafPage.setNextPage(newLeafPageId, tx, table);

			// change the prevPage pointer on the next page:

			PageID rightPageId;
			rightPageId = newLeafPage.getNextPage(tx, table);
			if (rightPageId.pageId != INVALID) {
				LeafPage rightPage;
				rightPage = new LeafPage(rightPageId, headerPage.getKeyType(tx,
						table), tx, table);

				rightPage.setPrevPage(newLeafPageId, tx, table);
				unpinPage(rightPageId, true /* = DIRTY */);

				// ASSERTIONS:
				// - newLeafPage, newLeafPageId valid and pinned
				// - currentLeafPage, currentLeafPageId valid and pinned
			}

			if (trace != null) {
				if (headerPage.getRootId(tx, table).pageId != currentLeafPageId.pageId)
					trace.writeBytes("SPLIT node " + currentLeafPageId
							+ " IN nodes " + currentLeafPageId + " "
							+ newLeafPageId + lineSep);
				else
					trace.writeBytes("ROOTSPLIT IN nodes " + currentLeafPageId
							+ " " + newLeafPageId + lineSep);
				trace.flush();
			}

			KeyDataEntry tmpEntry;
			RID firstRid = new RID();

			for (tmpEntry = currentLeafPage.getFirst(firstRid, tx, table); tmpEntry != null; tmpEntry = currentLeafPage
					.getFirst(firstRid, tx, table)) {

				newLeafPage.insertRecord(tmpEntry.key,
						((LeafData) (tmpEntry.data)).getData(), tx, table);
				currentLeafPage.deleteSortedRecord(firstRid, tx, table);

			}

			// ASSERTIONS:
			// - currentLeafPage empty
			// - newLeafPage holds all former records from currentLeafPage

			KeyDataEntry undoEntry = null;
			for (tmpEntry = newLeafPage.getFirst(firstRid, tx, table); newLeafPage
					.availableSpace(tx, table) < currentLeafPage
					.availableSpace(tx, table); tmpEntry = newLeafPage
					.getFirst(firstRid, tx, table)) {
				undoEntry = tmpEntry;
				// System.out.println("undoEntry:"+undoEntry.key);
				currentLeafPage.insertRecord(tmpEntry.key,
						((LeafData) tmpEntry.data).getData(), tx, table);
				newLeafPage.deleteSortedRecord(firstRid, tx, table);
			}

			// System.out.println("key:"+key);

			if (BT.KeyCompare(key, undoEntry.key) < 0) {
				// undo the final record

				if (currentLeafPage.availableSpace(tx, table) < newLeafPage
						.availableSpace(tx, table)) {
					newLeafPage.insertRecord(undoEntry.key,
							((LeafData) undoEntry.data).getData(), tx, table);

					currentLeafPage.deleteSortedRecord(new RID(currentLeafPage
							.getCurPage(tx, table), (int) currentLeafPage
							.getSlotCnt(tx, table) - 1), tx, table);
				}
			}

			// check whether <key, rid>
			// will be inserted
			// on the newly allocated or on the old leaf page

			if (BT.KeyCompare(key, undoEntry.key) >= 0) {
				// the new data entry belongs on the new Leaf page
				newLeafPage.insertRecord(key, rid, tx, table);

				if (trace != null) {
					trace.writeBytes("PUTIN node " + newLeafPageId + lineSep);
					trace.flush();
				}

			} else {
				currentLeafPage.insertRecord(key, rid, tx, table);
			}

			unpinPage(currentLeafPageId, true /* dirty */);

			if (trace != null) {
				trace_children(currentLeafPageId, tx, table);
				trace_children(newLeafPageId, tx, table);
			}

			// fill upEntry
			tmpEntry = newLeafPage.getFirst(firstRid, tx, table);
			upEntry = new KeyDataEntry(tmpEntry.key, newLeafPageId);

			unpinPage(newLeafPageId, true /* dirty */);

			// ASSERTIONS:
			// - no pages pinned
			// - upEntry holds the valid KeyDataEntry which is to be inserted
			// on the index page one level up
			return upEntry;
		} else {
			throw new BTInsertException("BT Insert Error");
		}
	}

	/**
	 * delete leaf entry given its <key, rid> pair. `rid' is IN the data entry;
	 * it is not the id of the data entry)
	 */

	public boolean Delete(KeyClass key, RID rid, Transaction tx, String table)
			throws Exception {
		if (headerPage.getDeleteFashion(tx, table) == DeleteFormat.FULL_DELETE)
			return FullDelete(key, rid, tx, table);
		else if (headerPage.getDeleteFashion(tx, table) == DeleteFormat.NAIVE_DELETE)
			return NaiveDelete(key, rid, tx, table);
		else
			throw new DeleteFormatException("Delete fashion was null!");
	}

	/**
	 * find left-most occurrence of `lo_key', going all the way left if lo_key
	 * is null.
	 * 
	 * Starting record returned in *pstartrid, on page *pppage, which is pinned.
	 * 
	 * Since we allow duplicates, this must "go left" as described in the text
	 * (for the search algorithm).
	 * 
	 * @param lo_key
	 *            find left-most occurrence of `lo_key', going all the way left
	 *            if lo_key is null.
	 * @param startrid
	 *            it will reurn the first rid =< lo_key
	 * @return return a BTLeafPage instance which is pinned. null if no key was
	 *         found.
	 */

	LeafPage findRunStart(KeyClass lo_key, RID startrid, Transaction tx,
			String table) throws Exception {
		LeafPage pageLeaf;
		IndexPage pageIndex;
		Page page;
		BTPage sortPage;
		PageID pageno;
		PageID prevpageno;
		PageID nextpageno;
		KeyDataEntry curEntry;

		pageno = headerPage.getRootId(tx, table);

		if (pageno.pageId == INVALID) { // no pages in the BTREE
			pageLeaf = null; // should be handled by
			// startrid =INVALID_PAGEID ; // the caller
			return pageLeaf;
		}

		page = pinPage(pageno);
		sortPage = new BTPage(page, headerPage.getKeyType(tx, table));

		if (trace != null) {
			trace.writeBytes("VISIT node " + pageno + lineSep);
			trace.flush();
		}

		// ASSERTION
		// - pageno and sortPage is the root of the btree
		// - pageno and sortPage valid and pinned

		while (sortPage.getType(tx, table) == NodeType.INDEX) {
			pageIndex = new IndexPage(page, headerPage.getKeyType(tx, table),
					tx, table);
			prevpageno = pageIndex.getPrevPage(tx, table);
			curEntry = pageIndex.getFirst(startrid, tx, table);
			while (curEntry != null && lo_key != null
					&& BT.KeyCompare(curEntry.key, lo_key) < 0) {

				prevpageno = ((IndexData) curEntry.data).getData();
				curEntry = pageIndex.getNext(startrid, tx, table);
			}

			unpinPage(pageno);

			pageno = prevpageno;
			page = pinPage(pageno);
			sortPage = new BTPage(page, headerPage.getKeyType(tx, table));

			if (trace != null) {
				trace.writeBytes("VISIT node " + pageno + lineSep);
				trace.flush();
			}

		}

		pageLeaf = new LeafPage(page, headerPage.getKeyType(tx, table), tx,
				table);

		curEntry = pageLeaf.getFirst(startrid, tx, table);
		while (curEntry == null) {
			// skip empty leaf pages off to left
			nextpageno = pageLeaf.getNextPage(tx, table);
			unpinPage(pageno);
			if (nextpageno.pageId == INVALID) {
				// oops, no more records, so set this scan to indicate this.
				return null;
			}

			pageno = nextpageno;
			pageLeaf = new LeafPage(pinPage(pageno), headerPage.getKeyType(tx,
					table), tx, table);
			curEntry = pageLeaf.getFirst(startrid, tx, table);
		}

		// ASSERTIONS:
		// - curkey, curRid: contain the first record on the
		// current leaf page (curkey its key, cur
		// - pageLeaf, pageno valid and pinned

		if (lo_key == null) {
			return pageLeaf;
			// note that pageno/pageLeaf is still pinned;
			// scan will unpin it when done
		}

		while (BT.KeyCompare(curEntry.key, lo_key) < 0) {
			curEntry = pageLeaf.getNext(startrid, tx, table);
			while (curEntry == null) { // have to go right
				nextpageno = pageLeaf.getNextPage(tx, table);
				unpinPage(pageno);

				if (nextpageno.pageId == INVALID) {
					return null;
				}

				pageno = nextpageno;
				pageLeaf = new LeafPage(pinPage(pageno), headerPage.getKeyType(
						tx, table), tx, table);

				curEntry = pageLeaf.getFirst(startrid, tx, table);
			}
		}
		
		return pageLeaf;
	}

	/*
	 * Status BTreeFile::NaiveDelete (const void *key, const RID rid)
	 * 
	 * Remove specified data entry (<key, rid>) from an index.
	 * 
	 * We don't do merging or redistribution, but do allow duplicates.
	 * 
	 * Page containing first occurrence of key `key' is found for us by
	 * findRunStart. We then iterate for (just a few) pages, if necesary, to
	 * find the one containing <key,rid>, which we then delete via
	 * BTLeafPage::delUserRid.
	 */

	private boolean NaiveDelete(KeyClass key, RID rid, Transaction tx,
			String table) throws Exception {
		LeafPage leafPage;
		RID curRid = new RID(); // iterator
		PageID nextpage;
		KeyDataEntry entry;

		if (trace != null) {
			trace.writeBytes("DELETE " + rid.pageNo + " " + rid.slotNo + " "
					+ key + lineSep);
			trace.writeBytes("DO" + lineSep);
			trace.writeBytes("SEARCH" + lineSep);
			trace.flush();
		}

		leafPage = findRunStart(key, curRid, tx, table); // find first
															// page,rid of key
		if (leafPage == null)
			return false;

		// System.out.println("leafPage:"+leafPage.getCurPage().pageId);
		entry = leafPage.getCurrent(curRid, tx, table);

		while (true) {

			while (entry == null) { // have to go right
				nextpage = leafPage.getNextPage(tx, table);
				unpinPage(leafPage.getCurPage(tx, table));
				if (nextpage.pageId == INVALID) {
					return false;
				}

				leafPage = new LeafPage(pinPage(nextpage), headerPage
						.getKeyType(tx, table), tx, table);
				entry = leafPage.getFirst(new RID(), tx, table);
			}

			if (BT.KeyCompare(key, entry.key) > 0)
				break;

			// System.out.println(entry.key);
			if (leafPage.delEntry(new KeyDataEntry(key, rid), tx, table) == true) {

				// successfully found <key, rid> on this page and deleted it.
				// unpin dirty page and return OK.
				unpinPage(leafPage.getCurPage(tx, table), true /* = DIRTY */);

				if (trace != null) {
					trace.writeBytes("TAKEFROM node "
							+ leafPage.getCurPage(tx, table) + lineSep);
					trace.writeBytes("DONE" + lineSep);
					trace.flush();
				}

				return true;
			}

			nextpage = leafPage.getNextPage(tx, table);
			unpinPage(leafPage.getCurPage(tx, table));

			// System.out.println(nextpage.pageId);
			leafPage = new LeafPage(pinPage(nextpage), headerPage.getKeyType(
					tx, table), tx, table);

			entry = leafPage.getFirst(curRid, tx, table);
		}

		/*
		 * We reached a page with first key > `key', so return an error. We
		 * should have got true back from delUserRid above. Apparently the
		 * specified <key,rid> data entry does not exist.
		 */

		unpinPage(leafPage.getCurPage(tx, table));
		return false;
	}

	private boolean FullDelete(KeyClass key, RID rid, Transaction tx,
			String table) throws Exception {
		if (trace != null) {
			trace.writeBytes("DELETE " + rid.pageNo + " " + rid.slotNo + " "
					+ key + lineSep);
			trace.writeBytes("DO" + lineSep);
			trace.writeBytes("SEARCH" + lineSep);
			trace.flush();
		}

		_Delete(key, rid, headerPage.getRootId(tx, table), null, tx, table);

		if (trace != null) {
			trace.writeBytes("DONE" + lineSep);
			trace.flush();
		}

		return true;
	}

	private KeyClass _Delete(KeyClass key, RID rid, PageID currentPageId,
			PageID parentPageId, Transaction tx, String table) throws Exception {

		BTPage sortPage;
		Page page;
		page = pinPage(currentPageId);
		sortPage = new BTPage(page, headerPage.getKeyType(tx, table));

		if (trace != null) {
			trace.writeBytes("VISIT node " + currentPageId + lineSep);
			trace.flush();
		}

		if (sortPage.getType(tx, table) == NodeType.LEAF) {
			RID curRid = new RID(); // iterator
			KeyDataEntry tmpEntry;
			KeyClass curkey;
			RID dummyRid;
			PageID nextpage;
			LeafPage leafPage;
			leafPage = new LeafPage(page, headerPage.getKeyType(tx, table), tx,
					table);

			KeyClass deletedKey = key;
			tmpEntry = leafPage.getFirst(curRid, tx, table);

			RID delRid;
			// for all records with key equal to 'key', delete it if its rid =
			// 'rid'
			while ((tmpEntry != null)
					&& (BT.KeyCompare(key, tmpEntry.key) >= 0)) {
				// WriteUpdateLog is done in the btleafpage level - to log the
				// deletion of the rid.

				if (leafPage.delEntry(new KeyDataEntry(key, rid), tx, table)) {
					// successfully found <key, rid> on this page and deleted
					// it.

					if (trace != null) {
						trace.writeBytes("TAKEFROM node "
								+ leafPage.getCurPage(tx, table) + lineSep);
						trace.flush();
					}

					PageID leafPage_no = leafPage.getCurPage(tx, table);
					if ((4 + leafPage.availableSpace(tx, table)) <= ((MAX_FRAME - HFPage.DPFIXED) / 2)) {
						// the leaf page is at least half full after the
						// deletion
						unpinPage(leafPage.getCurPage(tx, table), true /*
																		 * =
																		 * DIRTY
																		 */);
						return null;
					} else if (leafPage_no.pageId == headerPage.getRootId(tx,
							table).pageId) {
						// the tree has only one node - the root
						if (leafPage.numberOfRecords(tx, table) != 0) {
							unpinPage(leafPage_no, true /* = DIRTY */);
							return null;
						} else {
							// the whole tree is empty

							if (trace != null) {
								trace.writeBytes("DEALLOCATEROOT "
										+ leafPage_no + lineSep);
								trace.flush();
							}

							freePage(leafPage_no);

							updateHeader(new PageID(INVALID), tx, table);
							return null;
						}
					} else {
						// get a sibling
						IndexPage parentPage;
						parentPage = new IndexPage(pinPage(parentPageId),
								headerPage.getKeyType(tx, table), tx, table);

						PageID siblingPageId = new PageID();
						LeafPage siblingPage;
						int direction;
						direction = parentPage.getSibling(key, siblingPageId,
								tx, table);

						if (direction == 0) {
							// there is no sibling. nothing can be done.

							unpinPage(leafPage.getCurPage(tx, table), true /* =DIRTY */);

							unpinPage(parentPageId);

							return null;
						}

						siblingPage = new LeafPage(pinPage(siblingPageId),
								headerPage.getKeyType(tx, table), tx, table);

						if (siblingPage.redistribute(leafPage, parentPage,
								direction, deletedKey, tx, table)) {
							// the redistribution has been done successfully

							if (trace != null) {

								trace_children(leafPage.getCurPage(tx, table),
										tx, table);
								trace_children(siblingPage
										.getCurPage(tx, table), tx, table);

							}

							unpinPage(leafPage.getCurPage(tx, table), true);
							unpinPage(siblingPageId, true);
							unpinPage(parentPageId, true);
							return null;
						} else if ((siblingPage.availableSpace(tx, table) + 8 /* 2*sizeof(slot) */) >= ((MAX_FRAME - HFPage.DPFIXED) - leafPage
								.availableSpace(tx, table))) {

							// we can merge these two children
							// get old child entry in the parent first
							KeyDataEntry oldChildEntry;
							if (direction == -1)
								oldChildEntry = leafPage.getFirst(curRid, tx,
										table);
							// get a copy
							else {
								oldChildEntry = siblingPage.getFirst(curRid,
										tx, table);
							}

							// merge the two children
							LeafPage leftChild, rightChild;
							if (direction == -1) {
								leftChild = siblingPage;
								rightChild = leafPage;
							} else {
								leftChild = leafPage;
								rightChild = siblingPage;
							}

							// move all entries from rightChild to leftChild
							RID firstRid = new RID(), insertRid;
							for (tmpEntry = rightChild.getFirst(firstRid, tx,
									table); tmpEntry != null; tmpEntry = rightChild
									.getFirst(firstRid, tx, table)) {
								leftChild.insertRecord(tmpEntry, tx, table);
								rightChild.deleteSortedRecord(firstRid, tx,
										table);
							}

							// adjust chain
							leftChild.setNextPage(rightChild.getNextPage(tx,
									table), tx, table);
							if (rightChild.getNextPage(tx, table).pageId != INVALID) {
								LeafPage nextLeafPage = new LeafPage(rightChild
										.getNextPage(tx, table), headerPage
										.getKeyType(tx, table), tx, table);
								nextLeafPage.setPrevPage(leftChild.getCurPage(
										tx, table), tx, table);
								unpinPage(nextLeafPage.getCurPage(tx, table),
										true);
							}

							if (trace != null) {
								trace.writeBytes("MERGE nodes "
										+ leftChild.getCurPage(tx, table) + " "
										+ rightChild.getCurPage(tx, table)
										+ lineSep);
								trace.flush();
							}

							unpinPage(leftChild.getCurPage(tx, table), true);

							unpinPage(parentPageId, true);

							freePage(rightChild.getCurPage(tx, table));

							return oldChildEntry.key;
						} else {
							// It's a very rare case when we can do neither
							// redistribution nor merge.

							unpinPage(leafPage.getCurPage(tx, table), true);

							unpinPage(siblingPageId, true);

							unpinPage(parentPageId, true);

							return null;
						}
					} // get a sibling block
				}// delete success block

				nextpage = leafPage.getNextPage(tx, table);
				unpinPage(leafPage.getCurPage(tx, table));

				/*
				 * if (nextpage.pageId == INVALID) throw new
				 * RecordNotFoundException(null, "");
				 */

				leafPage = new LeafPage(pinPage(nextpage), headerPage
						.getKeyType(tx, table), tx, table);
				tmpEntry = leafPage.getFirst(curRid, tx, table);

			} // while loop

			/*
			 * We reached a page with first key > `key', so return an error. We
			 * should have got true back from delUserRid above. Apparently the
			 * specified <key,rid> data entry does not exist.
			 */

			unpinPage(leafPage.getCurPage(tx, table));
			// throw new RecordNotFoundException(null, "");
		}

		if (sortPage.getType(tx, table) == NodeType.INDEX) {
			PageID childPageId;
			IndexPage indexPage = new IndexPage(page, headerPage.getKeyType(tx,
					table), tx, table);
			childPageId = indexPage.getPageNoByKey(key, tx, table);

			// now unpin the page, recurse and then pin it again
			unpinPage(currentPageId);

			KeyClass oldChildKey = _Delete(key, rid, childPageId,
					currentPageId, tx, table);

			// two cases:
			// - oldChildKey == null: one level lower no merge has occurred:
			// - oldChildKey != null: one of the children has been deleted and
			// oldChildEntry is the entry to be deleted.

			indexPage = new IndexPage(pinPage(currentPageId), headerPage
					.getKeyType(tx, table), tx, table);

			if (oldChildKey == null) {
				unpinPage(indexPage.getCurPage(tx, table), true);
				return null;
			}

			// delete the oldChildKey

			// save possible old child entry before deletion
			PageID dummyPageId;
			KeyClass deletedKey = key;
			RID curRid = indexPage.deleteKey(oldChildKey, tx, table);

			if (indexPage.getCurPage(tx, table).pageId == headerPage.getRootId(
					tx, table).pageId) {
				// the index page is the root
				if (indexPage.numberOfRecords(tx, table) == 0) {
					BTPage childPage;
					childPage = new BTPage(indexPage.getPrevPage(tx, table),
							headerPage.getKeyType(tx, table));

					if (trace != null) {
						trace.writeBytes("CHANGEROOT from node "
								+ indexPage.getCurPage(tx, table) + " to node "
								+ indexPage.getPrevPage(tx, table) + lineSep);
						trace.flush();
					}

					updateHeader(indexPage.getPrevPage(tx, table), tx, table);
					unpinPage(childPage.getCurPage(tx, table));

					freePage(indexPage.getCurPage(tx, table));
					return null;
				}
				unpinPage(indexPage.getCurPage(tx, table), true);
				return null;
			}

			// now we know the current index page is not a root
			if ((4 /* sizeof slot */+ indexPage.availableSpace(tx, table)) <= ((MAX_FRAME - HFPage.DPFIXED) / 2)) {
				// the index page is at least half full after the deletion
				unpinPage(currentPageId, true);

				return null;
			} else {
				// get a sibling
				IndexPage parentPage;
				parentPage = new IndexPage(pinPage(parentPageId), headerPage
						.getKeyType(tx, table), tx, table);

				PageID siblingPageId = new PageID();
				IndexPage siblingPage;
				int direction;
				direction = parentPage
						.getSibling(key, siblingPageId, tx, table);
				if (direction == 0) {
					// there is no sibling. nothing can be done.

					unpinPage(indexPage.getCurPage(tx, table), true);

					unpinPage(parentPageId);

					return null;
				}

				siblingPage = new IndexPage(pinPage(siblingPageId), headerPage
						.getKeyType(tx, table), tx, table);

				int pushKeySize = 0;
				if (direction == 1) {
					pushKeySize = BT.getKeyLength(parentPage.findKey(
							siblingPage.getFirst(new RID(), tx, table).key, tx,
							table));
				} else if (direction == -1) {
					pushKeySize = BT.getKeyLength(parentPage.findKey(indexPage
							.getFirst(new RID(), tx, table).key, tx, table));
				}

				if (siblingPage.redistribute(indexPage, parentPage, direction,
						deletedKey, tx, table)) {
					// the redistribution has been done successfully

					if (trace != null) {

						trace_children(indexPage.getCurPage(tx, table), tx,
								table);
						trace_children(siblingPage.getCurPage(tx, table), tx,
								table);

					}

					unpinPage(indexPage.getCurPage(tx, table), true);

					unpinPage(siblingPageId, true);

					unpinPage(parentPageId, true);

					return null;
				} else if (siblingPage.availableSpace(tx, table) + 4 /*
																		 * slot
																		 * size
																		 */>= ((MAX_FRAME - HFPage.DPFIXED)
						- (indexPage.availableSpace(tx, table) + 4 /* slot size */)
						+ pushKeySize + 4 /* slot size */+ 4 /* pageId size */)) {

					// we can merge these two children

					// get old child entry in the parent first
					KeyClass oldChildEntry;
					if (direction == -1) {
						oldChildEntry = indexPage.getFirst(curRid, tx, table).key;
					} else {
						oldChildEntry = siblingPage.getFirst(curRid, tx, table).key;
					}

					// merge the two children
					IndexPage leftChild, rightChild;
					if (direction == -1) {
						leftChild = siblingPage;
						rightChild = indexPage;
					} else {
						leftChild = indexPage;
						rightChild = siblingPage;
					}

					if (trace != null) {
						trace.writeBytes("MERGE nodes "
								+ leftChild.getCurPage(tx, table) + " "
								+ rightChild.getCurPage(tx, table) + lineSep);
						trace.flush();
					}

					// pull down the entry in its parent node
					// and put it at the end of the left child
					RID firstRid = new RID(), insertRid;
					PageID curPageId;

					leftChild.insertKey(parentPage.findKey(oldChildEntry, tx,
							table), rightChild.getLeftLink(tx, table), tx,
							table);

					// move all entries from rightChild to leftChild
					for (KeyDataEntry tmpEntry = rightChild.getFirst(firstRid,
							tx, table); tmpEntry != null; tmpEntry = rightChild
							.getFirst(firstRid, tx, table)) {
						leftChild.insertKey(tmpEntry.key,
								((IndexData) tmpEntry.data).getData(), tx,
								table);
						rightChild.deleteSortedRecord(firstRid, tx, table);
					}

					unpinPage(leftChild.getCurPage(tx, table), true);

					unpinPage(parentPageId, true);

					freePage(rightChild.getCurPage(tx, table));

					return oldChildEntry; // ???

				} else {
					// It's a very rare case when we can do neither
					// redistribution nor merge.

					unpinPage(indexPage.getCurPage(tx, table), true);

					unpinPage(siblingPageId, true);

					unpinPage(parentPageId);

					return null;
				}
			}
		} // index node
		return null; // neither leaf and index page

	}

	public BTFileScan newScan(KeyClass lo_key, KeyClass hi_key,Transaction tx)
			throws Exception {
		BTFileScan scan = new BTFileScan();
		if (headerPage.getRootId(tx,dbname).pageId == INVALID) {
			scan.leafpage = null;
			return scan;
		}

		scan.btreefilename = dbname;
		scan.endkey = hi_key;
		scan.didfirst = false;
		scan.deletedcurrent = false;
		scan.curRid = new RID();
		scan.keyType = headerPage.getKeyType(tx,dbname);
		scan.maxKeysize = headerPage.getMaxKeySize(tx,dbname);
		scan.bfile = this;

		// this sets up scan at the starting position, ready for iteration
		scan.leafpage = findRunStart(lo_key, scan.curRid,tx,dbname);
		return scan;
	}

	void trace_children(PageID id, Transaction tx, String table)
			throws Exception {

		if (trace != null) {

			BTPage sortedPage;
			RID metaRid = new RID();
			PageID childPageId;
			KeyClass key;
			KeyDataEntry entry;
			sortedPage = new BTPage(pinPage(id), headerPage.getKeyType(tx,
					table));

			// Now print all the child nodes of the page.
			if (sortedPage.getType(tx, table) == NodeType.INDEX) {
				IndexPage indexPage = new IndexPage(sortedPage, headerPage
						.getKeyType(tx, table), tx, table);
				trace.writeBytes("INDEX CHILDREN " + id + " nodes" + lineSep);
				trace.writeBytes(" " + indexPage.getPrevPage(tx, table));
				for (entry = indexPage.getFirst(metaRid, tx, table); entry != null; entry = indexPage
						.getNext(metaRid, tx, table)) {
					trace
							.writeBytes("   "
									+ ((IndexData) entry.data).getData());
				}
			} else if (sortedPage.getType(tx, table) == NodeType.LEAF) {
				LeafPage leafPage = new LeafPage(sortedPage, headerPage
						.getKeyType(tx, table), tx, table);
				trace.writeBytes("LEAF CHILDREN " + id + " nodes" + lineSep);
				for (entry = leafPage.getFirst(metaRid, tx, table); entry != null; entry = leafPage
						.getNext(metaRid, tx, table)) {
					trace.writeBytes("   " + entry.key + " " + entry.data);
				}
			}
			unpinPage(id);
			trace.writeBytes(lineSep);
			trace.flush();
		}

	}
}
